﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using FinMkts.Messaging.Pgm;

namespace FinMkts.Pgm.ConsoleTest
{
    class Program
    {
        private Socket _socket;
        private int _count;
        private SocketAsyncEventArgs _args;
        private BlockBufferedBlockingQueue _queue;

        static void Main(string[] args)
        {
            new Program().Run();
        }

        private void Run()
        {
            var t = new Timer(TimerCallback);
            _socket = new Socket(AddressFamily.InterNetwork, SocketType.Rdm, RmConstants.ProtocolType);
            _socket.Bind(new IPEndPoint(IPAddress.Any, 0));

            //_socket.SetSocketOption(RmSocketOptionName.SetSendInterface, IPAddress.Parse("192.168.1.7"));
            _socket.SetSocketOption(RmSocketOptionName.SetTimeToLive, 1);
            _socket.SetSendWindow(new RmSendWindow{ RatePerSec = BinarySize.FromMegaBits(200), WindowLength = TimeSpan.FromSeconds(10)});

            var sendWindow = _socket.GetSendWindow();
            Console.WriteLine(sendWindow);

            // connect
            var complete = new AutoResetEvent(false);
            _args = new SocketAsyncEventArgs();
            _args.RemoteEndPoint = new IPEndPoint(IPAddress.Parse("234.5.6.7"), 40002);
            EventHandler<SocketAsyncEventArgs> connectCallback = (sender, eventArgs) => complete.Set();
            _args.Completed += connectCallback;
            _socket.ConnectAsync(_args);
            complete.WaitOne();
            Debug.Assert(_args.SocketError == SocketError.Success);
            _args.Completed -= connectCallback;

            t.Change(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(1));

            // send data via blocked queue
            _queue = new BlockBufferedBlockingQueue(BinarySize.FromKiloBytes(32), BlockReadyToSend);
            _args.Completed += SendAsyncCompleted;
            while (true)
            {
                byte[] message = Encoding.UTF8.GetBytes("so long, and thanks for all the fish");
                _queue.Add(message);
            }
            GC.KeepAlive(t);
        }

        private void SendAsyncCompleted(object sender, SocketAsyncEventArgs args)
        {
            if (args.SocketError != SocketError.Success)
            {
                throw new SocketException((int)_args.SocketError);
            }

            lock (_socket)
            {
                _count += args.BufferList.Count;
                BlockReadyToSend(_queue.TakeNextBlock());
            }
        }

        private void BlockReadyToSend(ByteBufferCollection buffers)
        {
            lock(_socket)
            {
                while (buffers != null)
                {
                    _args.BufferList = buffers;                   
                    _queue.IsBlocked = _socket.SendAsync(_args);
                    if (_args.SocketError != SocketError.Success)
                    {
                        throw new SocketException((int)_args.SocketError);
                    }
                    if (_queue.IsBlocked)
                    {
                        return;
                    }
                    _count += _args.BufferList.Count;
                    buffers = _queue.TakeNextBlock();
                }

                if (buffers == null)
                {
                    _queue.IsBlocked = false;
                }
            }
        }

        private void TimerCallback(object state)
        {
            int count;
            RmSenderStats stats;
            lock (_socket)
            {
                stats = _socket.GetSenderStats();
                count = _count;
                _count = 0;
            }
            Console.WriteLine("sent {0:N0} messages, rate {1}", count, stats.RatePerSecLast.ToString(BitsBytes.Bits));
        }
    }
}
